package com.li.flink.datastream

import com.li.flink.utils.MysqlUtils
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import java.sql.{Connection, PreparedStatement}

object SourceConnectMysql {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    var connection: Connection = null
    var prepareStmt: PreparedStatement = null
    //增强的并行RichParallelSourceFunction
    val sfRichParallel = new RichParallelSourceFunction[String] {
      override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = {
        prepareStmt = connection.prepareStatement("select * from student");
        val rs = prepareStmt.executeQuery();
        while (rs.next()) {
          val id = rs.getInt("id");
          val name = rs.getString("name")
          val age = rs.getInt("age")
          println(id, name, age)
        }
      }

      override def cancel(): Unit = {
        MysqlUtils.closeConnetion(connection)
      }

      override def open(parameters: Configuration): Unit = {
        connection = MysqlUtils.getConnection()
      }
    }
    env.addSource(sfRichParallel).print()
    env.execute("SourceConnectMysql")
  }
}